package com.joker.demo.admin.service.impl;


import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.joker.demo.admin.domain.constants.JobInfoConstants;
import com.joker.demo.admin.domain.po.ExecutorStatusPO;
import com.joker.demo.admin.domain.po.JobGroupPO;
import com.joker.demo.admin.domain.po.JobInfoPO;
import com.joker.demo.admin.domain.po.JobLogPO;
import com.joker.demo.admin.domain.po.JobRegistryPO;
import com.joker.demo.admin.domain.po.ThreadPoolMonitorPO;
import com.joker.demo.admin.domain.po.EnvironmentMonitorPO;
import com.joker.demo.admin.mapper.ExecutorStatusMapper;
import com.joker.demo.admin.mapper.JobGroupMapper;
import com.joker.demo.admin.mapper.JobInfoMapper;
import com.joker.demo.admin.mapper.JobLogMapper;
import com.joker.demo.admin.mapper.JobRegistryMapper;
import com.joker.demo.admin.service.JobInfoService;
import com.joker.demo.admin.service.ThreadPoolMonitorService;
import com.joker.demo.admin.service.EnvironmentMonitorService;
import com.joker.demo.admin.util.JobConvertUtils;
import com.joker.demo.admin.util.MapGetUtils;
import com.joker.demo.admin.util.ThreadPoolMonitorConvertUtils;
import com.joker.demo.admin.util.EnvironmentMonitorConvertUtils;
import com.joker.demo.core.domain.bo.JobRegistryBO;
import com.joker.demo.core.domain.bo.JobRegistryDetailBO;
import com.joker.demo.core.domain.dto.MessageDTO;
import com.joker.demo.core.domain.dto.TaskCancelRequestDTO;
import com.joker.demo.core.domain.dto.TaskExecuteRequestDTO;
import com.joker.demo.core.domain.dto.TaskExecuteResponseDTO;
import com.joker.demo.core.domain.dto.ThreadPoolMonitorDTO;
import com.joker.demo.core.domain.dto.EnvironmentMonitorDTO;
import com.joker.demo.core.domain.enums.JobBlockStrategyEnum;
import com.joker.demo.core.domain.enums.JobMisfireStrategyEnum;
import com.joker.demo.core.domain.enums.JobRoutePolicyEnum;
import com.joker.demo.core.domain.enums.JobRunStatusEnum;
import com.joker.demo.core.domain.enums.JobSchedulingTypeEnum;
import com.joker.demo.core.domain.enums.JobStatusEnum;
import com.joker.demo.core.domain.enums.JobTimeUnitEnum;
import com.joker.demo.core.domain.enums.JobRequestEnum;
import com.joker.demo.core.scheduler.JobScheduler;
import com.joker.demo.core.scheduler.JobSchedulerHandlerService;

import java.util.UUID;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 任务调度器
 *
 * @author : feixiang.li
 * @since : 2025-07-25 09:42
 */
@Slf4j
@Component
public class JobSchedulerHandlerServiceImpl implements JobSchedulerHandlerService {


    @Resource
    private JobInfoMapper jobInfoMapper;


    @Resource
    private JobInfoService jobInfoService;


    @Resource
    private JobGroupMapper jobGroupMapper;


    @Resource
    private JobRegistryMapper jobRegistryMapper;

    /**
     * 任务执行的记录
     */
    @Resource
    private JobLogMapper jobLogMapper;

    /**
     * 线程池监控服务
     */
    @Resource
    private ThreadPoolMonitorService threadPoolMonitorService;

    /**
     * 环境监控服务
     */
    @Resource
    private EnvironmentMonitorService environmentMonitorService;

    @Resource
    private ExecutorStatusMapper executorStatusMapper;

    @Resource
    private JobScheduler jobScheduler;


    private String getRegisterValue(JobRegistryBO jobRegistryBo) {
        if (Objects.isNull(jobRegistryBo) || CollUtil.isEmpty(jobRegistryBo.getJobRegistryList())) {
            return null;
        }
        String groupName = jobRegistryBo.getGroupName();
        List<String> jobNames = jobRegistryBo.getJobRegistryList().stream().map(item -> groupName + "-" + item.getJobName()).collect(Collectors.toList());
        return StringUtils.join(jobNames, ",");
    }

    /**
     * 注册任务处理器
     * 有几个部分需要处理
     *
     * @param jobRegistryBo 任务注册信息
     * @return Boolean
     */
    @Override
    public Boolean registerJobHandlers(JobRegistryBO jobRegistryBo) {

        if (Objects.isNull(jobRegistryBo) || CollUtil.isEmpty(jobRegistryBo.getJobRegistryList())) {
            log.info("注册任务处理器失败，参数为空");
            return false;
        }

        log.info("开始注册任务处理器:{}", JSON.toJSONString(jobRegistryBo));

        // 第一步，查找注册信息是否已经存在
        String groupName = jobRegistryBo.getGroupName();
        String registerKey = jobRegistryBo.getRegisterKey();

        JobRegistryPO jobRegistry = jobRegistryMapper.selectOne(new LambdaQueryWrapper<JobRegistryPO>().eq(JobRegistryPO::getRegistryKey, registerKey));

        String registryValue = getRegisterValue(jobRegistryBo);
        // 如果已经存在，则返回失败
        if (jobRegistry == null) {
            // 如果还未注册，更新3部分内容。第一部分。创建注册信息jobRegistry，第二部分。更新JobGroup内容，第三部分。更新任务jobInfo信息
            jobRegistry = new JobRegistryPO();
        }
        jobRegistry.setRegistryKey(registerKey);
        jobRegistry.setRegistryValue(registryValue);
        boolean insert = jobRegistryMapper.insertOrUpdate(jobRegistry);
        log.info("新增注册信息，数据:{},结果：{}", insert, jobRegistry);

        // 根据 groupName+jobName，变成唯一标识，来查找任务信息
        List<String> jobGroupNames = jobRegistryBo.getJobRegistryList().stream().map(item -> jobRegistryBo.getGroupName() + "-" + item.getJobName()).collect(Collectors.toList());

        // 第三步，根据唯一标识，查找任务信息
        List<JobInfoPO> jobInfoList = jobInfoMapper.selectList(new LambdaQueryWrapper<JobInfoPO>().in(JobInfoPO::getGroupJobName, jobGroupNames));

        Map<String, JobInfoPO> jobInfoMap = MapGetUtils.toMapWithFirstOccurrence(jobInfoList, JobInfoPO::getGroupJobName);

        // 从数据库中查找结果
        List<JobGroupPO> jobGroupList = jobGroupMapper.selectList(new LambdaQueryWrapper<JobGroupPO>().in(JobGroupPO::getGroupJobName, jobGroupNames));

        Map<String, JobGroupPO> jobGroupMap = MapGetUtils.toMapWithFirstOccurrence(jobGroupList, JobGroupPO::getGroupJobName);

        List<JobGroupPO> needUpdateGroupList = new ArrayList<>();
        List<JobGroupPO> addJobGroupList = new ArrayList<>();

        List<JobInfoPO> needUpdateJobInfoList = new ArrayList<>();
        List<JobInfoPO> addJobInfoList = new ArrayList<>();

        Date now = new Date();
        Date lastTriggle = DateUtils.addSeconds(now, JobInfoConstants.maxTriggerCount);
        for (JobRegistryDetailBO jobRegistryDetail : jobRegistryBo.getJobRegistryList()) {
            String groupJobName = groupName + "-" + jobRegistryDetail.getJobName();
            JobGroupPO jobGroup = jobGroupMap.get(groupJobName);
            // 如果不存在就新增，否则就更新
            if (jobGroup == null) {
                jobGroup = new JobGroupPO();
                jobGroup.setGroupJobName(groupJobName);
                jobGroup.setGroupName(groupName);
                jobGroup.setJobName(jobRegistryDetail.getJobName());
                jobGroup.setAddressList(jobRegistryBo.getRegisterKey());
                addJobGroupList.add(jobGroup);
            } else {
                // 如果存在，就更新状态
                jobGroup.addAddressList(jobRegistryBo.getRegisterKey());
                needUpdateGroupList.add(jobGroup);
            }

            // 将任务变成就绪状态
            JobInfoPO jobInfo = jobInfoMap.get(groupJobName);
            if (jobInfo == null) {
                jobInfo = new JobInfoPO();
                jobInfo.setGroupJobName(groupJobName);
                jobInfo.setGroupName(groupName);
                jobInfo.setJobName(jobRegistryDetail.getJobName());
                jobInfo.setJobDesc(jobRegistryDetail.getJobDesc());
                jobInfo.setCronExpression(jobRegistryDetail.getCronExpression());
                if (StringUtils.isBlank(jobRegistryDetail.getCronExpression())) {
                    jobInfo.setScheduleType(JobSchedulingTypeEnum.MANUAL.getCode());
                } else {
                    jobInfo.setScheduleType(JobSchedulingTypeEnum.CRON.getCode());
                }
                jobInfo.setSharding(jobRegistryDetail.getSharding());
                jobInfo.setMisfireStrategy(JobMisfireStrategyEnum.DO_NOTHING.getCode());
                jobInfo.setRouteStrategy(JobRoutePolicyEnum.DEFAULT.getCode());
                jobInfo.setBlockStrategy(JobBlockStrategyEnum.BLOCK.getCode());
                jobInfo.setRetryCount(jobRegistryDetail.getRetryCount());
                jobInfo.setTriggerStatus(JobStatusEnum.READY.getCode());
                // 设置超时时间
                jobInfo.setTimeout(JobTimeUnitEnum.getTimeOutToSecond(jobRegistryDetail.getTimeout(), jobRegistryDetail.getTimeoutUnit()));
                if (Objects.isNull(jobInfo.getTimeout()) || jobInfo.getTimeout() <= 0) {
                    jobInfo.setTimeout(30);
                }
                jobInfo.setAlarmEmail(jobRegistryDetail.getAlarmEmail());
                jobInfo.setLastTriggerTime(lastTriggle);
                JobConvertUtils.updateNextTriggerTime(jobInfo);
                addJobInfoList.add(jobInfo);
            } else {
                if (StringUtils.isBlank(jobRegistryDetail.getCronExpression())) {
                    jobInfo.setScheduleType(JobSchedulingTypeEnum.MANUAL.getCode());
                } else {
                    jobInfo.setScheduleType(JobSchedulingTypeEnum.CRON.getCode());
                }
                jobInfo.setCronExpression(jobRegistryDetail.getCronExpression());
                jobInfo.setSharding(jobRegistryDetail.getSharding());
                jobInfo.setJobDesc(jobRegistryDetail.getJobDesc());
                jobInfo.setTriggerStatus(JobStatusEnum.READY.getCode());
                jobInfo.setTimeout(JobTimeUnitEnum.getTimeOutToSecond(jobRegistryDetail.getTimeout(), jobRegistryDetail.getTimeoutUnit()));
                if (Objects.isNull(jobInfo.getTimeout()) || jobInfo.getTimeout() <= 0) {
                    jobInfo.setTimeout(30);
                }
                jobInfo.setAlarmEmail(jobRegistryDetail.getAlarmEmail());
                jobInfo.setRetryCount(jobRegistryDetail.getRetryCount());
                needUpdateJobInfoList.add(jobInfo);
            }
        }
        // 统一触发下一次触发时间
        // 更新JobGroup信息
        jobGroupMapper.insertOrUpdate(needUpdateGroupList);
        jobGroupMapper.insertOrUpdate(addJobGroupList);

        // 更新任务jobInfo信息
        jobInfoMapper.insertOrUpdate(needUpdateJobInfoList);
        jobInfoMapper.insertOrUpdate(addJobInfoList);

        return true;
    }

    /**
     * 下线任务处理器
     *
     * @param registerKey 应用名称
     * @return Boolean
     */
    @Override
    public Boolean unregisterJobHandlers(String registerKey) {
        log.info("【调度器】有执行器下线:{}", registerKey);
        // registerKey 为ip:port
        // 根据ip:port 查找对应的注册的应用
        JobRegistryPO jobRegistry = jobRegistryMapper.selectOne(new LambdaQueryWrapper<JobRegistryPO>().eq(JobRegistryPO::getRegistryKey, registerKey));
        if (jobRegistry == null) {
            log.info("下线任务处理器失败，参数为空");
            return false;
        }
        // 找到对应的任务应用
        List<String> groupJobNames = Arrays.asList(jobRegistry.getRegistryValue().split(","));

        List<String> needUpdateJobGroupNames = new ArrayList<>();
        List<JobGroupPO> jobGroupList = jobGroupMapper.selectList(new LambdaQueryWrapper<JobGroupPO>().in(JobGroupPO::getGroupJobName, groupJobNames));
        for (JobGroupPO jobGroup : jobGroupList) {
            Integer i = jobGroup.removeAddressList(registerKey);
            if (Objects.isNull(i) || i <= 0) {
                needUpdateJobGroupNames.add(jobGroup.getGroupJobName());
            }
        }
        // 更新任务应用信息
        jobGroupMapper.insertOrUpdate(jobGroupList);
        // 删除注册信息
        jobRegistryMapper.deleteById(jobRegistry);
        // 更新jobInfo状态-从就绪-> 异常
        if (CollUtil.isNotEmpty(needUpdateJobGroupNames)) {
            List<JobInfoPO> jobInfoList = jobInfoMapper.selectList(new LambdaQueryWrapper<JobInfoPO>().in(JobInfoPO::getGroupJobName, needUpdateJobGroupNames));
            for (JobInfoPO jobInfo : jobInfoList) {
                jobInfo.setTriggerStatus(JobStatusEnum.EXCEPTION.getCode());
            }
            jobInfoMapper.insertOrUpdate(jobInfoList);
        }
        return true;
    }

    /**
     * 调度器发送任务执行的请求EXECUTE_TASK,将任务更新成请求发送中。
     *
     * @param taskExecuteResponse 响应结果
     */
    @Override
    public void sendExecuteTask(TaskExecuteRequestDTO taskExecuteResponse) {
        log.info("调度器发送任务执行请求：{}", taskExecuteResponse);

        // 通过Netty Channel发送任务执行请求
        String registerKey = taskExecuteResponse.getRegisterKey();

        try {
            // 构建执行任务消息
            MessageDTO executeMessage = new MessageDTO(JobRequestEnum.EXECUTE_TASK.getCode(), UUID.randomUUID().toString(), JSON.toJSONString(taskExecuteResponse));

            // 通过Netty服务器发送消息到执行器
            boolean sendSuccess = jobScheduler.getNettyServer().sendMessageToExecutor(registerKey, executeMessage);

            if (!sendSuccess) {
                log.error("通过Netty发送任务执行请求失败，未找到执行器连接：{}", registerKey);
                return;
            }

            log.info("任务执行请求已通过Netty发送到执行器：{}", registerKey);

            // 更新任务状态
            JobLogPO jobLog = jobLogMapper.selectById(taskExecuteResponse.getJobLogId());
            if (Objects.isNull(jobLog)) {
                log.error("调度器发送任务执行的请求EXECUTE_TASK,将任务更新成请求发送中。发送失败:jobLogId：{}", taskExecuteResponse.getJobLogId());
                return;
            }
            jobLog.setTriggerCode(JobRunStatusEnum.RUNNING.getCode());
            jobLogMapper.updateById(jobLog);

        } catch (Exception e) {
            log.error("通过Netty发送任务执行请求失败，registerKey：{}，错误：{}", registerKey, e.getMessage(), e);
        }
    }

    /**
     * 调度器接收到EXECUTE_TASK_RESPONSE 请求，将任务变成 运行中
     *
     * @param taskExecuteResponse 任务执行结果
     */
    @Override
    public void execute(TaskExecuteResponseDTO taskExecuteResponse) {
        log.info("调度器接收任务执行结果：{}", taskExecuteResponse);
        JobLogPO jobLog = jobLogMapper.selectById(taskExecuteResponse.getJobLogId());
        if (jobLog == null) {
            log.error("调度器接收到EXECUTE_TASK_RESPONSE，但是任务不存在jobLogId：{}", taskExecuteResponse.getJobLogId());
            return;
        }
        jobLog.setTriggerCode(JobRunStatusEnum.RUNNING.getCode());
        jobLogMapper.updateById(jobLog);
    }

    /**
     * 执行器发送TASK_RESULT请求，将任务变成3种状态（运行失败/运行成功/运行超时）.
     *
     * @param taskExecuteResponse 任务执行结果
     */
    @Override
    public void receiveExecuteTaskResponse(TaskExecuteResponseDTO taskExecuteResponse) {
        log.info("【调度器】处理执行器发送任务结果请求：{}", taskExecuteResponse);
        JobLogPO jobLog = jobLogMapper.selectById(taskExecuteResponse.getJobLogId());
        if (jobLog == null) {
            log.error("【调度器】，但是任务不存在，jobLogId：{}", taskExecuteResponse.getJobLogId());
            return;
        }
        // TODO: 这里分为2种情况。第一种失败/第二种成功
        if (JobRunStatusEnum.FAILED.getCode().equals(taskExecuteResponse.getStatus())) {

            JobInfoPO jobInfo = jobInfoMapper.selectById(jobLog.getJobId());
            // 失败的话，需要判断是否重试/并且重试次数是否超过阈值。并发送告警信息
            if (jobLog.getExecutorFailRetryCount() < jobInfo.getRetryCount()) {
                // 构建新的任务立刻执行
            }
            // TODO:发送告警信息
        }

        jobLog.setTriggerCode(taskExecuteResponse.getStatus());
        jobLog.setTriggerMsg(taskExecuteResponse.getResult());
        jobLog.setStartTime(taskExecuteResponse.getStartTime());
        jobLog.setEndTime(taskExecuteResponse.getEndTime());
        jobLog.setTimeCost(jobLog.getEndTime().getTime() - jobLog.getStartTime().getTime());
        jobLogMapper.updateById(jobLog);
    }

    @Override
    public void sendCancelTask(TaskExecuteResponseDTO taskExecuteResponse) {

        // 构建任务取消请求
        TaskCancelRequestDTO cancelRequest = new TaskCancelRequestDTO();
        cancelRequest.setJobLogId(taskExecuteResponse.getJobLogId());

        // 通过Netty Channel发送任务取消请求
        String registerKey = taskExecuteResponse.getIp();

        try {
            // 构建取消任务消息
            MessageDTO cancelMessage = new MessageDTO(JobRequestEnum.CANCEL_TASK.getCode(), UUID.randomUUID().toString(), JSON.toJSONString(cancelRequest));

            // 通过Netty服务器发送消息到执行器
            boolean sendSuccess = jobScheduler.getNettyServer().sendMessageToExecutor(registerKey, cancelMessage);

            if (!sendSuccess) {
                log.error("通过Netty发送任务取消请求失败，未找到执行器连接：{}", registerKey);
                return;
            }

            log.info("任务取消请求已通过Netty发送到执行器：{}", registerKey);

            log.info("处理发送任务取消请求：{}", taskExecuteResponse);
            JobLogPO jobLog = jobLogMapper.selectById(taskExecuteResponse.getJobLogId());
            if (jobLog == null) {
                log.error("取消任务任务执行失败，任务不存在，jobLogId：{}", taskExecuteResponse.getJobLogId());
                return;
            }
            jobLog.setTriggerCode(JobRunStatusEnum.CANCELING.getCode());
            jobLogMapper.updateById(jobLog);

        } catch (Exception e) {
            log.error("通过Netty发送任务取消请求失败，registerKey：{}，错误：{}", registerKey, e.getMessage(), e);
        }
    }

    @Override
    public void cancel(TaskExecuteResponseDTO taskExecuteResponse) {
        log.info("处理任务取消响应：{}", taskExecuteResponse);
        JobLogPO jobLog = jobLogMapper.selectById(taskExecuteResponse.getJobLogId());
        if (jobLog == null) {
            log.error("任务执行失败，任务不存在，jobLogId：{}", taskExecuteResponse.getJobLogId());
            return;
        }
        jobLog.setTriggerCode(JobRunStatusEnum.CANCELED.getCode());
        jobLogMapper.updateById(jobLog);
    }

    @Override
    public void processThreadPoolMonitorData(ThreadPoolMonitorDTO monitorData) {
        if (monitorData == null) {
            log.warn("【调度器】接收到空的线程池监控数据");
            return;
        }

        try {
            log.info("【调度器】开始处理线程池监控数据，执行器：{}，应用：{}",
                    monitorData.getRegisterKey(), monitorData.getAppName());

            // 1. 数据验证
            if (!validateMonitorData(monitorData)) {
                log.warn("【调度器】线程池监控数据验证失败，执行器：{}", monitorData.getRegisterKey());
                return;
            }

            // 2. 转换并保存监控数据到数据库
            saveMonitorDataToDatabase(monitorData);

            // 3. 更新执行器状态信息
            updateExecutorStatus(monitorData);

            // 4. 检查告警条件
            checkAlarmConditions(monitorData);

            log.info("【调度器】线程池监控数据处理完成，执行器：{}", monitorData.getRegisterKey());

        } catch (Exception e) {
            log.error("【调度器】处理线程池监控数据失败，执行器：{}，错误：{}",
                    monitorData.getRegisterKey(), e.getMessage(), e);
        }
    }

    /**
     * 验证监控数据的有效性
     *
     * @param monitorData 监控数据
     * @return 验证结果
     */
    private boolean validateMonitorData(ThreadPoolMonitorDTO monitorData) {
        // 验证基本信息
        if (StringUtils.isBlank(monitorData.getRegisterKey()) ||
                StringUtils.isBlank(monitorData.getAppName())) {
            log.warn("监控数据基本信息不完整，registerKey：{}，appName：{}",
                    monitorData.getRegisterKey(), monitorData.getAppName());
            return false;
        }

        // 验证时间戳
        if (monitorData.getTimestamp() <= 0) {
            log.warn("监控数据时间戳无效：{}", monitorData.getTimestamp());
            return false;
        }

        // 验证线程池信息
        if (monitorData.getThreadPoolInfo() == null) {
            log.warn("监控数据缺少线程池信息");
            return false;
        }

        // 验证内存信息
        if (monitorData.getMemoryInfo() == null) {
            log.warn("监控数据缺少内存信息");
            return false;
        }

        return true;
    }

    /**
     * 保存监控数据到数据库
     *
     * @param monitorData 监控数据
     */
    private void saveMonitorDataToDatabase(ThreadPoolMonitorDTO monitorData) {
        try {
            // 转换DTO为PO
            ThreadPoolMonitorPO monitorPO = ThreadPoolMonitorConvertUtils.convertToPO(monitorData);
            if (monitorPO == null) {
                log.error("监控数据转换失败，执行器：{}", monitorData.getRegisterKey());
                return;
            }

            // 保存到数据库
            threadPoolMonitorService.saveMonitorData(monitorPO);
            log.debug("线程池监控数据保存成功，执行器：{}，时间：{}",
                    monitorData.getRegisterKey(),
                    new Date(monitorData.getTimestamp()));

        } catch (Exception e) {
            log.error("保存监控数据到数据库失败，执行器：{}，错误：{}",
                    monitorData.getRegisterKey(), e.getMessage(), e);
        }
    }

    /**
     * 更新执行器状态信息
     *
     * @param monitorData 监控数据
     */
    private void updateExecutorStatus(ThreadPoolMonitorDTO monitorData) {
        try {
            // 构建执行器状态数据
            ExecutorStatusPO executorStatus = buildExecutorStatus(monitorData);

            // 记录执行器状态更新日志
            logExecutorStatusUpdate(executorStatus);

            // 更新执行器状态
            LambdaQueryWrapper<ExecutorStatusPO> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(ExecutorStatusPO::getRegisterKey, executorStatus.getRegisterKey());
            ExecutorStatusPO existingStatus = executorStatusMapper.selectOne(queryWrapper);
            if (existingStatus != null) {
                executorStatus.setId(existingStatus.getId());
                executorStatusMapper.updateById(executorStatus);
            } else {
                executorStatusMapper.insert(executorStatus);
            }

        } catch (Exception e) {
            log.error("更新执行器状态失败，执行器：{}，错误：{}",
                    monitorData.getRegisterKey(), e.getMessage(), e);
        }
    }


    /**
     * 构建执行器状态对象
     *
     * @param monitorData 监控数据
     * @return 执行器状态对象
     */
    private ExecutorStatusPO buildExecutorStatus(ThreadPoolMonitorDTO monitorData) {
        ExecutorStatusPO executorStatus = new ExecutorStatusPO();

        // 设置基本信息
        executorStatus.setRegisterKey(monitorData.getRegisterKey());
        executorStatus.setAppName(monitorData.getAppName());
        executorStatus.setStatus("ONLINE");

        // 设置时间信息
        LocalDateTime monitorTime = LocalDateTime.ofInstant(
                new Date(monitorData.getTimestamp()).toInstant(),
                ZoneId.systemDefault());
        executorStatus.setLastMonitorTime(monitorTime);
        executorStatus.setLastHeartbeatTime(monitorTime);

        // 设置线程池信息
        if (monitorData.getThreadPoolInfo() != null) {
            ThreadPoolMonitorDTO.ThreadPoolInfo threadPoolInfo = monitorData.getThreadPoolInfo();
            executorStatus.setCurrentPoolSize(threadPoolInfo.getPoolSize());
            executorStatus.setCurrentActiveCount(threadPoolInfo.getActiveCount());
            executorStatus.setCurrentQueueSize(threadPoolInfo.getQueueSize());
        }

        // 设置内存信息
        if (monitorData.getMemoryInfo() != null) {
            executorStatus.setCurrentMemoryUsagePercent(
                    BigDecimal.valueOf(monitorData.getMemoryInfo().getMemoryUsagePercent()));
        }

        // 设置任务信息
        if (monitorData.getTaskStatistics() != null) {
            executorStatus.setCurrentRunningTasks(monitorData.getTaskStatistics().getRunningTaskCount());
        }

        return executorStatus;
    }

    /**
     * 记录执行器状态更新日志
     *
     * @param executorStatus 执行器状态
     */
    private void logExecutorStatusUpdate(ExecutorStatusPO executorStatus) {
        log.debug("执行器状态更新 - 执行器：{}，应用：{}，活跃线程数：{}，队列大小：{}，内存使用率：{}%，运行任务数：{}",
                executorStatus.getRegisterKey(),
                executorStatus.getAppName(),
                executorStatus.getCurrentActiveCount(),
                executorStatus.getCurrentQueueSize(),
                executorStatus.getCurrentMemoryUsagePercent(),
                executorStatus.getCurrentRunningTasks());
    }

    /**
     * 检查告警条件
     *
     * @param monitorData 监控数据
     */
    private void checkAlarmConditions(ThreadPoolMonitorDTO monitorData) {
        try {
            // 检查内存使用率告警
            checkMemoryUsageAlarm(monitorData);

            // 检查线程池队列积压告警
            checkQueueBacklogAlarm(monitorData);

            // 检查任务执行失败率告警
            checkTaskFailureRateAlarm(monitorData);

        } catch (Exception e) {
            log.error("检查告警条件失败，执行器：{}，错误：{}",
                    monitorData.getRegisterKey(), e.getMessage(), e);
        }
    }

    /**
     * 检查内存使用率告警
     *
     * @param monitorData 监控数据
     */
    private void checkMemoryUsageAlarm(ThreadPoolMonitorDTO monitorData) {
        if (monitorData.getMemoryInfo() != null) {
            double memoryUsagePercent = monitorData.getMemoryInfo().getMemoryUsagePercent();

            // 内存使用率超过80%告警
            if (memoryUsagePercent > 80.0) {
                log.warn("【内存告警】执行器：{}，内存使用率过高：{}%",
                        monitorData.getRegisterKey(), memoryUsagePercent);
                // TODO: 可以在这里添加告警通知逻辑
            }

            // 内存使用率超过90%严重告警
            if (memoryUsagePercent > 90.0) {
                log.error("【严重内存告警】执行器：{}，内存使用率严重过高：{}%",
                        monitorData.getRegisterKey(), memoryUsagePercent);
                // TODO: 可以在这里添加紧急告警通知逻辑
            }
        }
    }

    /**
     * 检查线程池队列积压告警
     *
     * @param monitorData 监控数据
     */
    private void checkQueueBacklogAlarm(ThreadPoolMonitorDTO monitorData) {
        if (monitorData.getThreadPoolInfo() != null) {
            int queueSize = monitorData.getThreadPoolInfo().getQueueSize();
            int queueCapacity = monitorData.getThreadPoolInfo().getQueueSize() +
                    monitorData.getThreadPoolInfo().getQueueRemainingCapacity();

            // 队列使用率超过80%告警
            if (queueCapacity > 0 && (double) queueSize / queueCapacity > 0.8) {
                log.warn("【队列告警】执行器：{}，队列积压严重，队列大小：{}/{}",
                        monitorData.getRegisterKey(), queueSize, queueCapacity);
                // TODO: 可以在这里添加告警通知逻辑
            }
        }
    }

    /**
     * 检查任务执行失败率告警
     *
     * @param monitorData 监控数据
     */
    private void checkTaskFailureRateAlarm(ThreadPoolMonitorDTO monitorData) {
        if (monitorData.getTaskStatistics() != null) {
            long completedTasks = monitorData.getTaskStatistics().getCompletedTasksLastMinute();
            long failedTasks = monitorData.getTaskStatistics().getFailedTasksLastMinute();

            // 计算失败率
            if (completedTasks > 0) {
                double failureRate = (double) failedTasks / completedTasks;

                // 失败率超过10%告警
                if (failureRate > 0.1) {
                    log.warn("【任务失败告警】执行器：{}，任务失败率过高：{}%，完成：{}，失败：{}",
                            monitorData.getRegisterKey(),
                            String.format("%.2f", failureRate * 100),
                            completedTasks, failedTasks);
                    // TODO: 可以在这里添加告警通知逻辑
                }
            }
        }
    }

    @Override
    public void processEnvironmentMonitorData(EnvironmentMonitorDTO monitorData) {
        if (monitorData == null) {
            log.warn("【调度器】接收到空的环境监控数据");
            return;
        }

        try {
            log.info("【调度器】开始处理环境监控数据，执行器：{}，应用：{}",
                    monitorData.getRegisterKey(), monitorData.getAppName());

            // 1. 数据验证
            if (!validateEnvironmentMonitorData(monitorData)) {
                log.warn("【调度器】环境监控数据验证失败，执行器：{}", monitorData.getRegisterKey());
                return;
            }

            // 2. 转换并保存监控数据到数据库
            saveEnvironmentMonitorDataToDatabase(monitorData);

            // 3. 检查告警条件
            checkEnvironmentAlarmConditions(monitorData);

            log.info("【调度器】环境监控数据处理完成，执行器：{}", monitorData.getRegisterKey());

        } catch (Exception e) {
            log.error("【调度器】处理环境监控数据失败，执行器：{}，错误：{}",
                    monitorData.getRegisterKey(), e.getMessage(), e);
        }
    }

    /**
     * 验证环境监控数据的有效性
     *
     * @param monitorData 环境监控数据
     * @return 验证结果
     */
    private boolean validateEnvironmentMonitorData(EnvironmentMonitorDTO monitorData) {
        // 验证基本信息
        if (StringUtils.isBlank(monitorData.getRegisterKey()) ||
                StringUtils.isBlank(monitorData.getAppName())) {
            log.warn("环境监控数据基本信息不完整，registerKey：{}，appName：{}",
                    monitorData.getRegisterKey(), monitorData.getAppName());
            return false;
        }

        // 验证时间戳
        if (monitorData.getTimestamp() <= 0) {
            log.warn("环境监控数据时间戳无效：{}", monitorData.getTimestamp());
            return false;
        }

        // 验证系统信息
        if (monitorData.getSystemInfo() == null) {
            log.warn("环境监控数据缺少系统信息");
            return false;
        }

        // 验证JVM信息
        if (monitorData.getJvmInfo() == null) {
            log.warn("环境监控数据缺少JVM信息");
            return false;
        }

        return true;
    }

    /**
     * 保存环境监控数据到数据库
     *
     * @param monitorData 环境监控数据
     */
    private void saveEnvironmentMonitorDataToDatabase(EnvironmentMonitorDTO monitorData) {
        try {
            // 转换DTO为PO
            EnvironmentMonitorPO monitorPO = EnvironmentMonitorConvertUtils.convertToPO(monitorData);
            if (monitorPO == null) {
                log.error("环境监控数据转换失败，执行器：{}", monitorData.getRegisterKey());
                return;
            }

            // 保存到数据库
            environmentMonitorService.saveMonitorData(monitorPO);
            log.debug("环境监控数据保存成功，执行器：{}，时间：{}",
                    monitorData.getRegisterKey(),
                    new Date(monitorData.getTimestamp()));

        } catch (Exception e) {
            log.error("保存环境监控数据到数据库失败，执行器：{}，错误：{}",
                    monitorData.getRegisterKey(), e.getMessage(), e);
        }
    }

    /**
     * 检查环境监控告警条件
     *
     * @param monitorData 环境监控数据
     */
    private void checkEnvironmentAlarmConditions(EnvironmentMonitorDTO monitorData) {
        try {
            // 检查系统内存使用率告警
            checkSystemMemoryUsageAlarm(monitorData);

            // 检查JVM堆内存使用率告警
            checkJvmHeapMemoryUsageAlarm(monitorData);

            // 检查磁盘使用率告警
            checkDiskUsageAlarm(monitorData);

            // 检查CPU使用率告警
            checkCpuUsageAlarm(monitorData);

        } catch (Exception e) {
            log.error("检查环境监控告警条件失败，执行器：{}，错误：{}",
                    monitorData.getRegisterKey(), e.getMessage(), e);
        }
    }

    /**
     * 检查系统内存使用率告警
     *
     * @param monitorData 环境监控数据
     */
    private void checkSystemMemoryUsageAlarm(EnvironmentMonitorDTO monitorData) {
        if (monitorData.getSystemInfo() != null) {
            Double systemMemoryUsagePercent = monitorData.getSystemInfo().getSystemMemoryUsagePercent();

            if (systemMemoryUsagePercent != null) {
                // 系统内存使用率超过80%告警
                if (systemMemoryUsagePercent > 80.0) {
                    log.warn("【系统内存告警】执行器：{}，系统内存使用率过高：{}%",
                            monitorData.getRegisterKey(), systemMemoryUsagePercent);
                    // TODO: 可以在这里添加告警通知逻辑
                }

                // 系统内存使用率超过90%严重告警
                if (systemMemoryUsagePercent > 90.0) {
                    log.error("【严重系统内存告警】执行器：{}，系统内存使用率严重过高：{}%",
                            monitorData.getRegisterKey(), systemMemoryUsagePercent);
                    // TODO: 可以在这里添加紧急告警通知逻辑
                }
            }
        }
    }

    /**
     * 检查JVM堆内存使用率告警
     *
     * @param monitorData 环境监控数据
     */
    private void checkJvmHeapMemoryUsageAlarm(EnvironmentMonitorDTO monitorData) {
        if (monitorData.getJvmInfo() != null) {
            Double heapMemoryUsagePercent = monitorData.getJvmInfo().getHeapMemoryUsagePercent();

            if (heapMemoryUsagePercent != null) {
                // JVM堆内存使用率超过80%告警
                if (heapMemoryUsagePercent > 80.0) {
                    log.warn("【JVM堆内存告警】执行器：{}，JVM堆内存使用率过高：{}%",
                            monitorData.getRegisterKey(), heapMemoryUsagePercent);
                    // TODO: 可以在这里添加告警通知逻辑
                }

                // JVM堆内存使用率超过90%严重告警
                if (heapMemoryUsagePercent > 90.0) {
                    log.error("【严重JVM堆内存告警】执行器：{}，JVM堆内存使用率严重过高：{}%",
                            monitorData.getRegisterKey(), heapMemoryUsagePercent);
                    // TODO: 可以在这里添加紧急告警通知逻辑
                }
            }
        }
    }

    /**
     * 检查磁盘使用率告警
     *
     * @param monitorData 环境监控数据
     */
    private void checkDiskUsageAlarm(EnvironmentMonitorDTO monitorData) {
        if (monitorData.getDiskInfos() != null) {
            for (EnvironmentMonitorDTO.DiskInfo diskInfo : monitorData.getDiskInfos()) {
                Double diskUsagePercent = diskInfo.getDiskUsagePercent();

                if (diskUsagePercent != null) {
                    // 磁盘使用率超过80%告警
                    if (diskUsagePercent > 80.0) {
                        log.warn("【磁盘告警】执行器：{}，磁盘：{}，使用率过高：{}%",
                                monitorData.getRegisterKey(), diskInfo.getDiskPath(), diskUsagePercent);
                        // TODO: 可以在这里添加告警通知逻辑
                    }

                    // 磁盘使用率超过90%严重告警
                    if (diskUsagePercent > 90.0) {
                        log.error("【严重磁盘告警】执行器：{}，磁盘：{}，使用率严重过高：{}%",
                                monitorData.getRegisterKey(), diskInfo.getDiskPath(), diskUsagePercent);
                        // TODO: 可以在这里添加紧急告警通知逻辑
                    }
                }
            }
        }
    }

    /**
     * 检查CPU使用率告警
     *
     * @param monitorData 环境监控数据
     */
    private void checkCpuUsageAlarm(EnvironmentMonitorDTO monitorData) {
        if (monitorData.getSystemInfo() != null) {
            Double cpuUsagePercent = monitorData.getSystemInfo().getCpuUsagePercent();

            if (cpuUsagePercent != null) {
                // CPU使用率超过80%告警
                if (cpuUsagePercent > 80.0) {
                    log.warn("【CPU告警】执行器：{}，CPU使用率过高：{}%",
                            monitorData.getRegisterKey(), cpuUsagePercent);
                    // TODO: 可以在这里添加告警通知逻辑
                }

                // CPU使用率超过90%严重告警
                if (cpuUsagePercent > 90.0) {
                    log.error("【严重CPU告警】执行器：{}，CPU使用率严重过高：{}%",
                            monitorData.getRegisterKey(), cpuUsagePercent);
                    // TODO: 可以在这里添加紧急告警通知逻辑
                }
            }
        }
    }
}
